package com._58city.spark.app.mr;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple2;

import com._58city.spark.app.DaoUtil;
import com._58city.spark.app.SparkUtil;
import com._58city.spark.app.dao.CPCDisplayDto;
import com._58city.spark.app.dao.CpcDisplayValue;
import com._58city.spark.app.ext.dto.DispCate;
import com._58city.spark.app.redis.Constant;
import com._58city.spark.app.redis.CpcDisplayModel;
import com._58city.spark.app.redis.RedisUtil;
import com._58city.spark.app.redis.TransferDimensionUtil;
import com._58city.spark.consumer.kafka.MessageAndMetadata;

public class MrKafkaCpcDisplay2 implements MrKafkaInterface2<String, CpcDisplayValue> {
	
	private static final long serialVersionUID = -8894846274742818214L;
	
	private Logger log = LoggerFactory.getLogger(MrKafkaCpcDisplay2.class);

	private Broadcast<Map<Long, DispCate>> bc_cate_map;

	private String[] keys;

	private int batchInterval;

	public MrKafkaCpcDisplay2(String[] keys, int batchInterval) {
		super();
		this.keys = keys;
		this.batchInterval = batchInterval;
	}

	public void setBc_cate_map(Broadcast<Map<Long, DispCate>> bc_cate_map) {
		this.bc_cate_map = bc_cate_map;
	}


	/**
	 * 将日志转化为一个CPC Display的Bean对象
	 * @param content
	 * @return
	 */
	private CPCDisplayDto toDto(MessageAndMetadata event) {
		if(null != event && null != event.getPayload()){
			String content = new String(event.getPayload());
			if (StringUtils.isNotBlank(content)) {
				CPCDisplayDto dto = DaoUtil.cpc_display(content);
				if (null == dto)
					return null;
				Map<Long, DispCate> cate_map = bc_cate_map.value();
				List<DispCate> cate_list = new ArrayList<DispCate>();
				for (String one : dto.getDisplayTypes()) {
					long cateid = Long.parseLong(one);
					DispCate dispCate = cate_map.get(cateid);
//					if (null == dispCate){
//						try {
//							dispCate = CacheUtil.getCateDict(cateid);
//						} catch (Exception e) {
//							log.info("CPC Display toDto cate error");
//						}
//					}
					if (null != dispCate)
						cate_list.add(dispCate);
				}
				dto.setBusiLine(cate_list.size() > 0 ? cate_list.get(0)
						.getBusiness_type_biz() : "other");
				return dto;
			}
		}
		return null;
	}

	public List<JavaPairDStream<String, CpcDisplayValue>> mapPair(
			List<JavaDStream<MessageAndMetadata>> kafkaStreams) {
		List<JavaPairDStream<String, CpcDisplayValue>> list = new ArrayList<JavaPairDStream<String, CpcDisplayValue>>();
		for (JavaDStream<MessageAndMetadata> stream : kafkaStreams) {
			JavaPairDStream<String, CpcDisplayValue> mapStream = stream
					.mapToPair(new PairFunction<MessageAndMetadata, String, CpcDisplayValue>() {
						private static final long serialVersionUID = -1151252686518724680L;

						public Tuple2<String, CpcDisplayValue> call(
								MessageAndMetadata event) throws Exception {
							try {
								CPCDisplayDto dto = toDto(event); 
								if (null != dto) {
									String key = SparkUtil.getMapKeyNotTime(dto, keys); //通过反射获取一个key
									if (null != key) {
										long pv = 1;
										long adNum = dto.getAdNum();
										long adPv = 0;
										if (adNum > 0)
											adPv = 1;
										CpcDisplayValue value = new CpcDisplayValue( //统计需要的值
												pv, adPv, adNum);
										return new Tuple2<String, CpcDisplayValue>( //K,V结构用于存储到RDD
												key, value);
									}
								}
							} catch (Exception e) {
								log.error("MrCpcDisplay map ERROR!", e);
							}
							return new Tuple2<String, CpcDisplayValue>(
									SparkUtil.ERROR, new CpcDisplayValue());
						}
					});
			list.add(mapStream);
		}
		return list;
	}

	public JavaPairDStream<String, CpcDisplayValue> reducePair(
			JavaPairDStream<String, CpcDisplayValue> mapStream) {
		JavaPairDStream<String, CpcDisplayValue> reducePair = mapStream
				.reduceByKey(new Function2<CpcDisplayValue, CpcDisplayValue, CpcDisplayValue>() {
					private static final long serialVersionUID = 1035857556638312606L;

					/*
					 * 将相同Key的V值，相加
					 */
					public CpcDisplayValue call(CpcDisplayValue v1,
							CpcDisplayValue v2) throws Exception {
						try {
							long pv = v1.getPv() + v2.getPv();
							long adNum = v1.getAdNum() + v2.getAdNum();
							long adPv = v1.getAdPv() + v2.getAdPv();
							return new CpcDisplayValue(pv, adPv, adNum);
						} catch (Exception e) {
							log.error("MrCpcDisplay reduce ERROR!", e);
						}
						return new CpcDisplayValue();
					}
				});
		return reducePair;
	}

//	public void foreachRDD(JavaPairDStream<String, CpcDisplayValue> reduceStream) {
//		reduceStream
//		.foreachRDD(new Function<JavaPairRDD<String, CpcDisplayValue>, Void>() {
//			private static final long serialVersionUID = 7874315786186348784L;
//
//			public Void call(JavaPairRDD<String, CpcDisplayValue> v)
//					throws Exception {
//				long now = new Date().getTime();
//				final long current_time = SparkUtil.makeFormateDate(
//						now, batchInterval);
//
//				v.foreach(new VoidFunction<Tuple2<String, CpcDisplayValue>>() {
//					private static final long serialVersionUID = -7618371311981204035L;
//
//					public void call(
//							Tuple2<String, CpcDisplayValue> tuple)
//							throws Exception {
//						try {
//							if (null != tuple) {
//								String keyStr = tuple._1;
//								if (!SparkUtil.ERROR.equals(keyStr)) {
//									CpcDisplayValue value = tuple._2;
//									String[] key_values = keyStr
//											.split(",");
//
//									long pv = value.getPv();
//									long adNum = value.getAdNum();
//									long adPv = value.getAdPv();
//
//									log.info("CPC Display ========== key : " + keyStr + ",time : " + current_time
//											+ ",pv : " + pv + ",adPv : " + adPv + ",adNum : " + adNum);
//
//									CpcDisplayModel cdm = new CpcDisplayModel();
//									cdm.setTime(current_time);
//									TransferDimensionUtil.dimension(cdm, keys, Integer.parseInt(key_values[0]), key_values[1]);
//									cdm.setPv(pv);
//									cdm.setAdPv(adPv);
//									cdm.setAdNum(adNum);
//									try {
//										RedisUtil.cache.lpushString(Constant.CPC_DISPLAY_REDIS_QUEUE_KEY,
//														JsonUtil.toJson(cdm)); 
//									} catch (Exception e) {
//										log.error("display redis insert error!");
//									}
//								}
//							}
//						} catch (Exception e) {
//							log.error("MrCpcDisplay foreachRDD ERROR!",
//									e);
//						}
//					}
//				});
//				return null;
//			}
//		});
//		
//	}
	
	public void foreachRDD(JavaPairDStream<String, CpcDisplayValue> reduceStream) {
		reduceStream
		.foreachRDD(new Function<JavaPairRDD<String, CpcDisplayValue>, Void>() {
			private static final long serialVersionUID = 7874315786186348784L;

			public Void call(JavaPairRDD<String, CpcDisplayValue> v)
					throws Exception {
				long now = new Date().getTime();
				final long current_time = SparkUtil.makeFormateDate(
						now, batchInterval);

				v.foreachPartition(new VoidFunction<Iterator<Tuple2<String,CpcDisplayValue>>>() {

					private static final long serialVersionUID = -6044909612721613780L;

					public void call(Iterator<Tuple2<String, CpcDisplayValue>> iter)
							throws Exception {
						if(null != iter){
							final List<CpcDisplayModel> list = new ArrayList<CpcDisplayModel>();
							while(iter.hasNext()){
								 Tuple2<String, CpcDisplayValue> tuple = iter.next();
								 String keyStr = tuple._1;
								 if (!SparkUtil.ERROR.equals(keyStr)) {
									 CpcDisplayValue value = tuple._2;
									 String[] key_values = keyStr.split(",");
	                                 long pv = value.getPv();
									 long adNum = value.getAdNum();
								  	 long adPv = value.getAdPv();
	                                 log.info("CPC Display ========== key : " + keyStr + 
	                                		  ",time : " + current_time+ 
	                                		  ",pv : " + pv + ",adPv : " + adPv +
	                                		  ",adNum : " + adNum);

									 CpcDisplayModel cdm = new CpcDisplayModel();
									 cdm.setTime(current_time);
									 TransferDimensionUtil.dimension(cdm, keys, Integer.parseInt(key_values[0]), key_values[1]);
									 cdm.setPv(pv);
									 cdm.setAdPv(adPv);
									 cdm.setAdNum(adNum);
									 list.add(cdm);
								 }
						    }
							
							if(list.size() > 0){
								Executors.newCachedThreadPool().submit(new Runnable() {
									public void run() {
                                        RedisUtil.send(Constant.CPC_DISPLAY_REDIS_QUEUE_KEY, list);										
									}
								});
							}
						}
					}
				});
				return null;
			}
		});
		
	}

	

}
